package com.shujia.gma.dim


import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo2SkuInfoDim {
  def main(args: Array[String]): Unit = {

    val fsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //计划器
      .inStreamingMode()
      .build()

    val fsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val fsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(fsEnv, fsSettings)


    fsTableEnv.executeSql(
      """
        |CREATE TABLE ods_base_category1 (
        |  id BIGINT,
        |  name STRING
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'gma.base_category1',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup',
        | 'format' = 'canal-json',
        | 'scan.startup.mode' = 'earliest-offset',
        | 'canal-json.ignore-parse-errors'='true'
        |)
        |
        |
      """.stripMargin)

    fsTableEnv.executeSql(
      """
        |CREATE TABLE ods_base_category2 (
        |  id BIGINT,
        |  name STRING,
        |  category1_id BIGINT
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'gma.base_category2',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup',
        | 'format' = 'canal-json',
        | 'scan.startup.mode' = 'earliest-offset',
        | 'canal-json.ignore-parse-errors'='true'
        |)
        |
        |
      """.stripMargin)

    fsTableEnv.executeSql(
      """
        |CREATE TABLE ods_base_category3 (
        |  id BIGINT,
        |  name STRING,
        |  category2_id BIGINT
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'gma.base_category3',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup',
        | 'format' = 'canal-json',
        | 'scan.startup.mode' = 'earliest-offset',
        | 'canal-json.ignore-parse-errors'='true'
        |)
        |
        |
      """.stripMargin)


    fsTableEnv.executeSql(
      """
        |CREATE TABLE ods_base_trademark (
        |  tm_id STRING,
        |  tm_name STRING
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'gma.base_trademark',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup',
        | 'format' = 'canal-json',
        | 'scan.startup.mode' = 'earliest-offset',
        | 'canal-json.ignore-parse-errors'='true'
        |)
        |
        |
      """.stripMargin)


    fsTableEnv.executeSql(
      """
        |CREATE TABLE ods_sku_info (
        | `id` bigint,
        |`spu_id` bigint,
        |`price` decimal(10,0),
        |`sku_name` String,
        |`sku_desc` String,
        |`weight` decimal(10,2),
        |`tm_id` bigint,
        |`category3_id` bigint,
        |`sku_default_img` String,
        |`create_time` TIMESTAMP(3)
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'gma.sku_info',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup',
        | 'format' = 'canal-json',
        | 'scan.startup.mode' = 'earliest-offset',
        | 'canal-json.ignore-parse-errors'='true'
        |)
        |
        |
      """.stripMargin)


    /**
      * create 'dim:sku_info','i'
      *
      */
    fsTableEnv.executeSql(
      """
        |CREATE TABLE dim_sku_info (
        | id BIGINT,
        | i ROW<spu_id bigint,price decimal(10,0),sku_name STRING,sku_desc STRING,weight decimal(10,2),tm_name string,base_category3_name STRING,base_category2_name STRING ,base_category1_name STRING,sku_default_img string,create_time TIMESTAMP(3)>,
        | PRIMARY KEY (id) NOT ENFORCED
        |) WITH (
        | 'connector' = 'hbase-1.4',
        | 'table-name' = 'dim:sku_info',
        | 'zookeeper.quorum' = 'master:2181,node1,2181,node2:2181',
        | 'sink.buffer-flush.max-rows'='1'
        |)
        |
      """.stripMargin)


    fsTableEnv.executeSql(
      """
        |
        |insert into dim_sku_info
        |select id,ROW(spu_id,price,sku_name,sku_desc,weight,tm_name,base_category3_name,base_category2_name,base_category1_name,sku_default_img,create_time) from
        |(select
        | a.id,a.spu_id,a.price,a.sku_name,a.sku_desc,a.weight,
        | b.tm_name,
        | c.name as base_category3_name,
        | d.name as base_category2_name,
        | e.name as base_category1_name,
        | a.sku_default_img,
        | a.create_time
        | from
        |ods_sku_info as a
        |join ods_base_trademark as b on cast(a.tm_id as String)=b.tm_id
        |join ods_base_category3 as c on a.category3_id=c.id
        |join ods_base_category2 as d on c.category2_id=d.id
        |join ods_base_category1 as e on d.category1_id=e.id
        |) as a
      """.stripMargin)


  }
}
